package com.aotain.project.apollo.boltv2;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * @ClassName: DNSAttackBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年11月8日 
 *
 */
public class DNSAttackBolt extends AbstractBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 固定源60s内握手成功次数  目标|源|端口
	 */
	private HashMap<String,Integer> fixedSourceMap = new HashMap<String,Integer>();

	/**
	 * 伪造源60s内握手成功次数  目标|端口
	 */
	private HashMap<String,Integer> forgedSourceMap = new HashMap<String,Integer>();

	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> upStreamoctets = new HashMap<String,Long>();

	/**
	 * 上行包数
	 */
	private HashMap<String,Long> upStreampacket = new HashMap<String,Long>();

	/**
	 * 受固定源攻击的记录map
	 */
	private HashMap<String,String> abnormalStreamFixedMap = new HashMap<String,String>();

	/**
	 * 受伪造源攻击的记录map
	 */
	private HashMap<String,String> abnormalStreamForgedMap = new HashMap<String,String>();

	/**
	 * 攻击开始时间
	 */
	private HashMap<String,Long> attackStartMap = new HashMap<String,Long>();

	/**
	 * 攻击持续的上次更新时间
	 */
	private HashMap<String,Long> attackLasttimeMap = new HashMap<String,Long>();

	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				Logger.getLogger(SYNFloodBolt.class).info(String.format("###DEBUG:DNSATTACK START EMIT [%s]",tuple.getSourceStreamId()));
				emitCountingData(collector);
				Logger.getLogger(SYNFloodBolt.class).info(String.format("###DEBUG:DNSATTACK END EMIT [%s]",tuple.getSourceStreamId()));
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("DNSATTACKBolt throws an exception!",e);
		}
	}


	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{

			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idc = tuple.getStringByField("idcno");
			String type = tuple.getStringByField("type");
			Integer status = tuple.getIntegerByField("status");
			String gis =  new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
			String dport = tuple.getStringByField("dport");
			long upStreamOctets = tuple.getLongByField("upstreamoctets");
			long upStreamPacket = tuple.getLongByField("upstreampacket");

			String[] gisArray = gis.split("#");								
			String destAreaName = gisArray[0];
			String destGis = gisArray[1];

			int fixedSourceCount = 1;
			int forgedSourceCount = 1;
			long upStreamOctetsFixed = upStreamOctets;
			long upStreamOctetsForged = upStreamOctets;
			long upStreamPacketFixed = upStreamPacket;
			long upStreamPacketForged = upStreamPacket;

			String fixedMapKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc + "|isFixed";
			String forgedMapKey = dip + "|" + dport + "|" + idc + "|" + destAreaName + "|" + destGis;
			//如果是udp 且目的端口为53 则计数
			if("53".equals(dport) && "17".equals(type)) {

				if(fixedSourceMap.containsKey(fixedMapKey)) {
					fixedSourceCount = fixedSourceMap.get(fixedMapKey) + 1;
				}

				if(forgedSourceMap.containsKey(forgedMapKey)) {
					forgedSourceCount = forgedSourceMap.get(forgedMapKey) + 1;
				}

				if(upStreamoctets.containsKey(fixedMapKey)) {
					upStreamOctetsFixed += upStreamoctets.get(fixedMapKey) ;
				}

				if(upStreamoctets.containsKey(forgedMapKey)) {
					upStreamOctetsForged += upStreamoctets.get(forgedMapKey);
				}
				if(upStreampacket.containsKey(fixedMapKey)) {
					upStreamPacketFixed += upStreampacket.get(fixedMapKey);
				}
				if(upStreampacket.containsKey(forgedMapKey)) {
					upStreamPacketForged += upStreampacket.get(forgedMapKey);
				}
				fixedSourceMap.put(fixedMapKey, fixedSourceCount);
				forgedSourceMap.put(forgedMapKey, forgedSourceCount);
				upStreamoctets.put(fixedMapKey, upStreamOctetsFixed);
				upStreamoctets.put(forgedMapKey, upStreamOctetsForged);
				upStreampacket.put(fixedMapKey, upStreamPacketFixed);
				upStreampacket.put(forgedMapKey, upStreamPacketForged);
			}


		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}

	public void emitCountingData(OutputCollector collector) {
		try {
			//		InsertIntoData();

			//固定源
			for(Entry<String, Integer> entry : fixedSourceMap.entrySet()) {

				String entryKey = entry.getKey();
				long currenttime = System.currentTimeMillis();
				
				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];
//				if("183.62.140.140".equals(dip))				
//					System.out.println("DNSATTACK---Fixed----attnum=" + upStreampacket.get(entryKey)/60);
				
				if(upStreampacket.get(entryKey) > 600000) {//如果每秒包数大于10000则判断有攻击

					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry,true);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,true,currenttime);
					}
				}else {
					//如果没大于门限 也时间超过3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {

							EndIntoData(entryKey,true);
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
						}
					}
				}
			}

			//伪造源攻击判断
			for(Entry<String, Integer> entry : forgedSourceMap.entrySet()) {

				String entryKey = entry.getKey();
				long currenttime = System.currentTimeMillis();
				
				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];
				
				if(upStreampacket.get(entryKey) > 600000) {//如果每秒包数大于10000则判断有攻击

					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry,false);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,false,currenttime);
					}
				}else {
					//如果没大于门限 也时间超过3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {

							EndIntoData(entryKey,false);
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
						}
					}
				}
			}

			// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束，避免漏处理数据
			ArrayList<String> removeItem = new ArrayList<String>();
			for(Entry<String,Long> entry:attackLasttimeMap.entrySet()) {
				String entryKey = entry.getKey();
				if(System.currentTimeMillis() - entry.getValue() > END_DURATION) {
					removeItem.add(entryKey);
					if(entryKey.endsWith("isFixed")) {
						EndIntoData(entryKey,true);
					} else {
						EndIntoData(entryKey,false);
					}
					attackStartMap.remove(entryKey);
				}
			}

			for(String key  : removeItem) {
				attackLasttimeMap.remove(key);
			}

			removeItem.clear();
			upStreamoctets.clear();
			upStreampacket.clear();
			fixedSourceMap.clear();
			forgedSourceMap.clear();
		} catch (Exception e ) {
			e.printStackTrace();
			Logger.getLogger(getClass()).error("DNS Exception",e);;
		}
	}



	/**
	 * 新记录入库
	 * @Title: InsertIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	//	public void InsertIntoData() {
	public void InsertIntoData(Entry<String,Integer> entry,boolean isFixed,long time) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);

		String destAreaName = "";
		String destGis = "";
		String sourceAreaName = "";
		String sourceGis = "";
		String sourceAreaCountry = "";
		String sourceAreaId = "";		
		String sourceProvinceName =  "";; //如果省为空，精确到国家
		String sourceAreaCityId = "";
		String sourceAreaProvinceId = "";
		String dip = "";
		String sip = "";
		String dport = "";
		String gis = "";
		String idc = "";
		String desc = "";
		String rowKeyAbnormal = "";

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(time));
		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		String rowkey3 = String.format("%s_%s", accesstime, sip);

		//KEY
		String entryKey = entry.getKey();

		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[2];
			gis = tuplesp[3];
			idc = tuplesp[4];
			String[] gisArray = gis.split("#");								
			destAreaName = gisArray[0];
			destGis = gisArray[1];
			sourceAreaName = gisArray[2];
			sourceGis = gisArray[3];
			sourceAreaCountry = gisArray[4];
			sourceAreaId = gisArray[5];		
			sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
			sourceAreaCityId = gisArray[7];
			sourceAreaProvinceId = gisArray[8];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_"  + dport + "_11";	
			desc = String.format("DNS固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", upStreampacket.get(entryKey), upStreamoctets.get(entryKey)/(1024*60)); 
		} else {
			dip = tuplesp[0];
			sip = "伪造源";
			dport = tuplesp[1];
			idc = tuplesp[2];				
			destAreaName = tuplesp[3];;
			destGis = tuplesp[4];
			sourceAreaName = "伪造源";
			sourceAreaCountry = "伪造源";
			sourceAreaId = "伪造源";	
			sourceProvinceName = "伪造源"; //如果省为空，精确到国家
			sourceAreaCityId = "伪造源";
			sourceAreaProvinceId = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_11";
			desc = String.format("DNS伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", upStreampacket.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));
		}
		
		System.out.println("DNSATTACK----InsertIntoData" + rowKeyAbnormal  + " : " + desc);

		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", dport);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", "11");
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));

		hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
		hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
		hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
		//			    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", "");
		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "2");
		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);

		Date current = new Date();
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);

		collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				sourceProvinceName, 1l, dateStr_hour,"DNSATTACK", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)upStreampacket.get(entryKey)));


	}

	/**
	 * 更新记录字段
	 * @Title: UpdateData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void UpdateData(Entry<String,Integer> entry,boolean isFixed) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);

		String dip = "";
		String sip = "";
		String desc = "";
		String dport = "";				
		String rowKeyAbnormal = "";

		String entryKey = entry.getKey();
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		//		String rowkey3 = String.format("%s_%s", accesstime, sip);

		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[2];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_11";	
			desc = String.format("DNSATTACK固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", upStreampacket.get(entryKey), upStreamoctets.get(entryKey)/(1024*60)); 
		} else {
			dip = tuplesp[0];
			dport = tuplesp[1];
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_" + dport + "_11";
			desc = String.format("DNSATTACK伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", upStreampacket.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));
		}
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String idc = tuplesp[4];
		
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);

		Date current = new Date();
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
	    String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);
		
		collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				sourceProvinceName, 0l, dateStr_hour,"DNSATTACK", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)upStreampacket.get(entryKey)));

	}

	/**
	 * 攻击结束记录入库
	 * @Title: EndIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param entryKey
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void EndIntoData(String entryKey,boolean isFixed) {
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);

		String dip = "";
		String sip = "";
		String desc = "";
		String dport = "";
		String rowKeyAbnormal = "";

		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
		//		String rowkey3 = String.format("%s_%s", accesstime, sip);


		String[] tuplesp = entryKey.split("\\|",-1);
		if(isFixed) {
			dip = tuplesp[0];
			sip = tuplesp[1];
			dport = tuplesp[2];
			rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_11";	
			desc = String.format("DNSATTACK固定源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000); 
		} else {
			dip = tuplesp[0];
			dport = tuplesp[1];
			sip = "伪造源";
			rowKeyAbnormal = dip + "_" + accesstime + "_ANY_"+ dport  + "_11";
			desc = String.format("DNSATTACK伪造源攻击结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000);
		}
		
		System.out.println("DNSATTACK----EndIntoData" + rowKeyAbnormal  + " : " + desc);
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
//		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ENDTIME", new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackLasttimeMap.get(entryKey)+60*1000l)));

	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour",
				"abntype","idcno","sourcecountry","streamoctets","streampacket"));
		//		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport",
		//				"accesstime","sourcearea","sourcegeo","sourcecountry","destarea","destgeo",
		//				"evaluate","abnormal","desc","attnum","delaytime","flag"));

	}

	@Override
	public Map<String, Object> getComponentConfiguration() { 
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60); //60秒统计一次
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub

	}

//	/**
//	 * 
//	 */
//	private static final long serialVersionUID = 1L;
//
//	/**
//	 * 固定源60s内握手成功次数  目标|源|端口
//	 */
//	private HashMap<String,Integer> fixedSourceMap = new HashMap<String,Integer>();
//
//	/**
//	 * 伪造源60s内握手成功次数  目标|端口
//	 */
//	private HashMap<String,Integer> forgedSourceMap = new HashMap<String,Integer>();
//
//	/**
//	 * 上行包字节数
//	 */
//	private HashMap<String,Integer> upStreamoctets = new HashMap<String,Integer>();
//
//	/**
//	 * 上行包数
//	 */
//	private HashMap<String,Integer> upStreampacket = new HashMap<String,Integer>();
//
//	/**
//	 * 受固定源攻击的记录map
//	 */
//	private HashMap<String,String> abnormalStreamFixedMap = new HashMap<String,String>();
//
//	/**
//	 * 受伪造源攻击的记录map
//	 */
//	private HashMap<String,String> abnormalStreamForgedMap = new HashMap<String,String>();
//
//
//	@Override
//	public void execute(Tuple tuple) {
//		// TODO Auto-generated method stub
//		try {
//			if(isTickTuple(tuple)) {
//				Logger.getLogger(DNSAttackBolt.class).info(String.format("###DEBUG:DNSAttack START EMIT [%s]",tuple.getSourceStreamId()));
//				emitCountingData(collector);
//				Logger.getLogger(DNSAttackBolt.class).info(String.format("###DEBUG:DNSAttack END EMIT [%s]",tuple.getSourceStreamId()));
//				collector.ack(tuple);
//			} else {
//				countInLocal(tuple); 
//				collector.ack(tuple);
//			}
//		} catch (Exception e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//			collector.fail(tuple);
//			throw new FailedException("DNSAttackBolt throws an exception!",e);
//		}
//	}
//
//	private boolean isTickTuple(Tuple tuple) {
//		return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
//				Constants.SYSTEM_TICK_STREAM_ID);   
//	}
//
//	//计算需要的值
//	private void countInLocal(Tuple tuple) {
//		try
//		{
//
//			String dip = tuple.getStringByField("dip");
//			String sip = tuple.getStringByField("sip");
//			String idc = tuple.getStringByField("idcno");
//			String type = tuple.getStringByField("type");
//			Integer status = tuple.getIntegerByField("status");
//			String gis =  new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
//			String dport = tuple.getStringByField("dport");
//			Integer upStreamOctets = tuple.getIntegerByField("upstreamoctets");
//			Integer upStreamPacket = tuple.getIntegerByField("upstreampacket");
//
//			String[] gisArray = gis.split("#");								
//			String destAreaName = gisArray[0];
//			String destGis = gisArray[1];
//
//			int fixedSourceCount = 1;
//			int forgedSourceCount = 1;
//			Integer upStreamOctetsFixed = upStreamOctets;
//			Integer upStreamOctetsForged = upStreamOctets;
//			Integer upStreamPacketFixed = upStreamPacket;
//			Integer upStreamPacketForged = upStreamPacket;
//			
//			String fixedMapKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc;
//			String forgedMapKey = dip + "|" + dport + "|" + idc + "|" + destAreaName + "|" + destGis;
//			//如果是udp 且目的端口为53 则计数
//			if("53".equals(dport) && "11".equals(type)) {
//
//				if(fixedSourceMap.containsKey(fixedMapKey)) {
//					fixedSourceCount = fixedSourceMap.get(fixedMapKey) + 1;
//				}
//
//				if(forgedSourceMap.containsKey(forgedMapKey)) {
//					forgedSourceCount = forgedSourceMap.get(forgedMapKey) + 1;
//				}
//
//				if(upStreamoctets.containsKey(fixedMapKey)) {
//					upStreamOctetsFixed += upStreamoctets.get(fixedMapKey) ;
//				}
//
//				if(upStreamoctets.containsKey(forgedMapKey)) {
//					upStreamOctetsForged += upStreamoctets.get(forgedMapKey);
//				}
//				if(upStreampacket.containsKey(fixedMapKey)) {
//					upStreamPacketFixed += upStreampacket.get(fixedMapKey);
//				}
//				if(upStreampacket.containsKey(forgedMapKey)) {
//					upStreamPacketForged += upStreampacket.get(forgedMapKey);
//				}
//			}
//
//			fixedSourceMap.put(fixedMapKey, fixedSourceCount);
//			forgedSourceMap.put(forgedMapKey, forgedSourceCount);
//			upStreamoctets.put(fixedMapKey, upStreamOctetsFixed);
//			upStreamoctets.put(forgedMapKey, upStreamOctetsForged);
//			upStreampacket.put(fixedMapKey, upStreamPacketFixed);
//			upStreampacket.put(forgedMapKey, upStreamPacketForged);
//			
//		}
//		catch(Exception ex)
//		{
//			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
//		}
//	}
//
//	private void emitCountingData(OutputCollector collector) {
//
//		String destAreaName = "";
//		String destGis = "";
//		String sourceAreaName = "";
//		String sourceGis = "";
//		String sourceAreaCountry = "";
//		String sourceAreaId = "";		
//		String sourceProvinceName =  "";; //如果省为空，精确到国家
//		String sourceAreaCityId = "";
//		String sourceAreaProvinceId = "";
//		Date current = new Date();
//		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
//		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
//		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
//		String dateStr_h = sdf_h.format(current);
//		String dateStr_d = sdf_d.format(current);
//		String dateStr_hour = sdf_hour.format(current);
//		boolean isFixed = false ;//是否为固定源
//
//		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
//				zooserver);
//		try
//		{
//
//
//			SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");		
//
//			for(Entry<String, Integer> entry : fixedSourceMap.entrySet())
//			{
//				//KEY
//				String entryKey = entry.getKey();
//				String[] tuplesp = entryKey.split("\\|",-1);
//				String dip = tuplesp[0];
//				String sip = tuplesp[1];
//				String dport = tuplesp[2];
//				String gis = tuplesp[3];
//				String idc = tuplesp[4];
//
//				String[] gisArray = gis.split("#");								
//				destAreaName = gisArray[0];
//				destGis = gisArray[1];
//				sourceAreaName = gisArray[2];
//				sourceGis = gisArray[3];
//				sourceAreaCountry = gisArray[4];
//				sourceAreaId = gisArray[5];		
//				sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
//				sourceAreaCityId = gisArray[7];
//				sourceAreaProvinceId = gisArray[8];
//
//				String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
//				String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
//
//				String abnMapKey = dip + "|" + sip + "|" + dport;
//				String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "_11";
//				String rowkey3 = String.format("%s_%s", accesstime, sip);
//
//				if(fixedSourceMap.get(entryKey) > 600000) {
//					isFixed = true;
//					String desc = String.format("DNS固定源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entry.getValue(), upStreamoctets.get(entryKey)/(1024*60));
//					
//					if(!abnormalStreamForgedMap.containsKey(abnMapKey)) {
//						hbaseInstance.Add("abnormalStreamForgedMap", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
//						abnormalStreamFixedMap.put(abnMapKey, accesstime);
//					} else {
//						rowKeyAbnormal = dip + "_" + abnormalStreamFixedMap.get(abnMapKey) + "_" + sip + "_" + "_11";//如果是还在持续的攻击，则只是更新记录
//					}
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", "");
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", "11");
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", entry.getValue().toString());
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
//
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", entry.getValue().toString());
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
//					//			    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", "");
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "2");
//					hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
//
//
//				} else {
//					if(abnormalStreamForgedMap.containsKey(abnMapKey)) {
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ENDTIME", accesstime);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "ENDTIME", accesstime);
//						abnormalStreamFixedMap.remove(abnMapKey);
//					}
//					isFixed = false;
//				}
//
//
//				collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
//						sourceProvinceName, entry.getValue(), dateStr_hour,"DNS", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)upStreampacket.get(entryKey)));
//
//			}
//
//			if(!isFixed) {
//				for(Entry<String, Integer> entry : forgedSourceMap.entrySet())
//				{
//					//KEY
//					String entryKey = entry.getKey();
//					String[] tuplesp = entryKey.split("\\|",-1);
//					String dip = tuplesp[0];
//					String sip = "伪造源";
//					String dport = tuplesp[1];
//					String idc = tuplesp[2];				
//					destAreaName = tuplesp[3];;
//					destGis = tuplesp[4];
//					sourceAreaName = "伪造源";
//					//				sourceGis = "伪造源";
//					sourceAreaCountry = "伪造源";
//					sourceAreaId = "伪造源";	
//					sourceProvinceName = "伪造源"; //如果省为空，精确到国家
//					sourceAreaCityId = "伪造源";
//					sourceAreaProvinceId = "伪造源";
//					String forged4Key = dip + "|" + dport + "|" + idc + "|" + destAreaName + "|" + destGis;
//
//					String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
//					String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
//
//					String abnForgedMapKey = dip + "|" + dport;
//					String rowKeyAbnormalFixed = dip + "_" + accesstime + "" + "_11";
//					String rowkey3 = String.format("%s_%s", accesstime, sip);
//
//					if(forgedSourceMap.get(entryKey) > 600000) {
//
//						String desc = String.format("DNS伪造源攻击 ,攻击次数[%d],攻击峰值流量[%d KB/s]", entry.getValue(), upStreamoctets.get(entryKey)/(1024*60));
//
//						if(!abnormalStreamForgedMap.containsKey(abnForgedMapKey)) {
//							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "ACCESSTIME", accesstime);
//							abnormalStreamForgedMap.put(abnForgedMapKey, accesstime);
//						}else {
//							rowKeyAbnormalFixed = dip + "_" + abnormalStreamFixedMap.get(abnForgedMapKey) + "_ANY" + "_11";//如果是还在持续的攻击，则只是更新记录
//						}
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "SOURCEIP", sip);
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "DESTPORT", "");
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "ABRNORMAL", "11");
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "DESTIP", dip);							
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "SOURCEAREA", sourceAreaName);
//						//					hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "SOURCEGEO", sourceGis);
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "SOURCECOUNTRY", sourceAreaCountry);
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "DESTAREA", destAreaName);
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "DESTGEO", destGis);							
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "ATTNUM", entry.getValue().toString());
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "DESC", desc);
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "EVALUATE", "40");
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "PROVINCE", sourceProvinceName);
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
//						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
//
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", entry.getValue().toString());
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
//						//			    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", "");
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "2");
//						hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
//
//
//					} else {
//						if(abnormalStreamForgedMap.containsKey(abnForgedMapKey)) {
//							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormalFixed, "cf", "ENDTIME", accesstime);
//							hbaseInstance.Add(tbName3, rowkey3, "cf", "ENDTIME", accesstime);
//							abnormalStreamFixedMap.remove(abnForgedMapKey);
//						}
//					}
//
//
//					collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
//							sourceProvinceName, entry.getValue(), dateStr_hour,"DNS", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)upStreampacket.get(entryKey)));
//					
//				}
//			}
//
//		} catch (Exception e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//			throw new FailedException("DNSAttackBolt emitCountingData throws an exception!",e);
//		} finally {
//			fixedSourceMap.clear();
//			forgedSourceMap.clear();
//			upStreamoctets.clear();
//			upStreampacket.clear();
//		}
//	}
//
//
//
//	@Override
//	public void cleanup() {
//		// TODO Auto-generated method stub
//
//	}
//
//	@Override
//	public void declareOutputFields(OutputFieldsDeclarer declarer) {
//		// TODO Auto-generated method stub
//		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d",
//				"dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour",
//				"abntype","idcno","sourcecountry","streamoctets","streampacket"));
//		//		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport",
//		//				"accesstime","sourcearea","sourcegeo","sourcecountry","destarea","destgeo",
//		//				"evaluate","abnormal","desc","attnum","delaytime","flag"));
//
//	}
//
//	@Override
//	public Map<String, Object> getComponentConfiguration() {
//		// TODO Auto-generated method stub
//		Map<String,Object> conf = new HashMap<String,Object>();
//		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60); //10秒统计一次
//		return conf;
//	}
//
//
//	@Override
//	public void Init(Map stormConf, TopologyContext context,
//			OutputCollector collector) {
//		// TODO Auto-generated method stub
//
//	}
}
